=============== 配置集群 ==============
mkdir /usr/kafka
cd /usr/kafka
wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz

vi server.properties
//三个节点都要根据不同的ip 设置broker.id=0 1 2
broker.id=0
zookeeper.connect=192.168.75.151:2181,192.168.75.152:2181,192.168.75.153:2181,192.168.75.154:2181,192.168.75.155:2181
log.dirs==/tmp/kafka-logs # 数据存储地址


# 配置环境变量
sudo tee  > /etc/profile.d/kafka_env.sh <<-'EOF'
export KFK_HOME=/usr/kafka/kafka_2.13-2.8.0
export PATH=$PATH:$KFK_HOME/bin
EOF

#启动
 ./bin/kafka-server-start.sh -daemon ./config/server.properties
#停止
./bin/kafka-server-stop.sh
 
# 配置ACL SCRAM-SHA-512 模式

# server kafka_server_jaas.conf
# config/kafka_server_jaas.conf
KafkaServer {
     org.apache.kafka.common.security.scram.ScramLoginModule required
     username="admin"
     password="admin";
};

#修改bin/kafka-server-start.sh

#exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=$KFK_HOME/config/kafka_server_jaas.conf kafka.Kafka "$@"

#重启之前配置用户
kafka-configs.sh --zookeeper 192.168.75.151:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin

# server.properties
# 启用ACL
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 设置本例中admin为超级用户
super.users=User:admin
# 启用SCRAM机制，采用SCRAM-SHA-512算法
sasl.enabled.mechanisms=SCRAM-SHA-512
# 为broker间通讯开启SCRAM机制，采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
# broker间通讯使用PLAINTEXT，本例中不演示SSL配置
security.inter.broker.protocol=SASL_PLAINTEXT
# 配置listeners使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://192.168.75.151:9092
# 配置advertised.listeners 外网访问地址
#advertised.listeners=SASL_PLAINTEXT://172.21.0.9:9092

# 配置kafka-server-start.sh 添加kafka_server_jaas.conf启动
# 配置consumer.properties和producer.properties
# jps 
=============== 创建主题 ==============
# 测试 --zookeeper localhost:2181
kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092

bin/kafka-topics.sh --create --topic cyc-location --replication-factor 2 --partitions 12 --zookeeper localhost:2181
# 查看主题
./bin/kafka-topics.sh --describe cyc-location --zookeeper localhost:2181
./bin/kafka-topics.sh --list --zookeeper localhost:2181



# 创建用户 zookeeper
kafka-configs.sh --zookeeper 192.168.75.151:2181 --alter --add-config 'SCRAM-SHA-256=[password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name tjhk_writer
kafka-configs.sh --zookeeper 192.168.75.151:2181 --alter --add-config 'SCRAM-SHA-256=[password=123456],SCRAM-SHA-512=[password=123456]' --entity-type users --entity-name tjhk_reader
# 查看所有用户
kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --list
# 查看用户权限
kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name tjhk_reader
# 授权
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:tjhk_writer --operation Write --topic test-20210715 --group test-group
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:tjhk_reader --operation Read --topic test-20210715
或 推荐上面方法
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice   --producer  --topic test
./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:alice   --consumer  --group=*  --topic test

# 生产 消费
kafka-console-producer.sh --broker-list 192.168.75.151:9092 --topic test-20210715 --producer.config tjhk_writer_jaas.conf
kafka-console-consumer.sh --bootstrap-server 192.168.75.151:9092 --topic test-20210715 --from-beginning --consumer.config tjhk_reader_jaas.conf --group test-group

